/*
 * Created on Apr 1, 2008
 * Created by Paul Gardner
 * 
 * Copyright (C) Azureus Software, Inc, All Rights Reserved.
 * 
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; version 2 of the License only.
 * 
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * GNU General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307, USA.
 */

package com.aelitis.azureus.plugins.net.buddy;

import java.io.File;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.URL;
import java.net.URLEncoder;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;

import org.gudy.azureus2.core3.util.AERunnable;
import org.gudy.azureus2.core3.util.AESemaphore;
import org.gudy.azureus2.core3.util.AEThread2;
import org.gudy.azureus2.core3.util.AddressUtils;
import org.gudy.azureus2.core3.util.BDecoder;
import org.gudy.azureus2.core3.util.BEncoder;
import org.gudy.azureus2.core3.util.Base32;
import org.gudy.azureus2.core3.util.Debug;
import org.gudy.azureus2.core3.util.DelayedEvent;
import org.gudy.azureus2.core3.util.LightHashMap;
import org.gudy.azureus2.core3.util.RandomUtils;
import org.gudy.azureus2.core3.util.SystemTime;
import org.gudy.azureus2.plugins.messaging.MessageException;
import org.gudy.azureus2.plugins.messaging.generic.GenericMessageConnection;
import org.gudy.azureus2.plugins.messaging.generic.GenericMessageConnectionListener;
import org.gudy.azureus2.plugins.messaging.generic.GenericMessageEndpoint;
import org.gudy.azureus2.plugins.messaging.generic.GenericMessageRegistration;
import org.gudy.azureus2.plugins.utils.PooledByteBuffer;
import org.gudy.azureus2.plugins.utils.security.SEPublicKey;
import org.gudy.azureus2.plugins.utils.security.SEPublicKeyLocator;
import org.gudy.azureus2.plugins.utils.security.SESecurityManager;

import com.aelitis.azureus.core.util.AZ3Functions;

public class BuddyPluginBuddy {
    private static final boolean TRACE = BuddyPlugin.TRACE;

    private static final int CONNECTION_IDLE_TIMEOUT = 5 * 60 * 1000;
    private static final int CONNECTION_KEEP_ALIVE = 1 * 60 * 1000;

    private static final int MAX_ACTIVE_CONNECTIONS = 5;
    private static final int MAX_QUEUED_MESSAGES = 256;

    private static final int RT_REQUEST_DATA = 1;

    private static final int RT_REPLY_DATA = 2;
    private static final int RT_REPLY_ERROR = 99;

    private BuddyPlugin plugin;
    private long created_time;
    private int subsystem;
    private boolean authorised;
    private String public_key;
    private String nick_name;
    private List<Long> recent_ygm;

    private int last_status_seq;

    private long post_time;
    private InetAddress ip;
    private int tcp_port;
    private int udp_port;
    private int online_status = BuddyPlugin.STATUS_ONLINE; // default

    private int version = BuddyPlugin.VERSION_CHAT; // assume everyone now supports chat

    private boolean online;
    private long last_time_online;

    private long status_check_count;
    private long last_status_check_time;

    private boolean check_active;

    private List<buddyConnection> connections = new ArrayList<buddyConnection>();
    private List<buddyMessage> messages = new ArrayList<buddyMessage>();
    private buddyMessage current_message;

    private int next_connection_id;
    private int next_message_id;

    private boolean ygm_active;
    private boolean ygm_pending;

    private long latest_ygm_time;
    private String last_message_received;

    private Set<Long> offline_seq_set;

    private int message_out_count;
    private int message_in_count;
    private int message_out_bytes;
    private int message_in_bytes;

    private String received_frag_details = "";

    private BuddyPluginBuddyMessageHandler persistent_msg_handler;

    private Map<Object, Object> user_data = new LightHashMap<Object, Object>();

    private boolean keep_alive_outstanding;
    private volatile long last_connect_attempt = SystemTime.getCurrentTime();
    private volatile int consec_connect_fails;

    private long last_auto_reconnect = -1;

    private Object rss_lock = new Object();

    private Set<String> rss_local_cats;
    private Set<String> rss_remote_cats;
    private Set<String> rss_cats_read;

    private AESemaphore outgoing_connect_sem = new AESemaphore("BPB:outcon", 1);

    private volatile boolean closing;
    private volatile boolean destroyed;

    protected BuddyPluginBuddy(BuddyPlugin _plugin, long _created_time, int _subsystem, boolean _authorised, String _pk, String _nick_name,
            int _version, String _rss_local_cats, String _rss_remote_cats, int _last_status_seq, long _last_time_online, List<Long> _recent_ygm) {
        plugin = _plugin;
        created_time = _created_time;
        subsystem = _subsystem;
        authorised = _authorised;
        public_key = _pk;
        nick_name = _nick_name;
        version = Math.max(version, _version);
        rss_local_cats = stringToCats(_rss_local_cats);
        rss_remote_cats = stringToCats(_rss_remote_cats);
        last_status_seq = _last_status_seq;
        last_time_online = _last_time_online;
        recent_ygm = _recent_ygm;

        persistent_msg_handler = new BuddyPluginBuddyMessageHandler(this, new File(plugin.getBuddyConfigDir(), public_key));
    }

    protected void setInitialStatus(long now, int num_buddies) {
        // for inactive buddies we schedule their status checks so that on average we don't
        // do more than one check every 5 minutes

        if (last_time_online == 0 && now - created_time > 7 * 24 * 60 * 60 * 1000L) {

            last_status_check_time = now + RandomUtils.nextInt(5 * 60 * 1000 * num_buddies);
        }
    }

    protected BuddyPlugin getPlugin() {
        return (plugin);
    }

    public BuddyPluginBuddyMessageHandler getMessageHandler() {
        return (persistent_msg_handler);
    }

    protected void persistentDispatchPending() {
        plugin.persistentDispatchPending(this);
    }

    protected void checkPersistentDispatch() {
        persistent_msg_handler.checkPersistentDispatch();
    }

    protected void persistentDispatch() {
        persistent_msg_handler.persistentDispatch();
    }

    public Map readConfigFile(File name) {
        return (plugin.readConfigFile(name));
    }

    public boolean writeConfigFile(File name, Map data) {
        return (plugin.writeConfigFile(name, data));
    }

    protected long getCreatedTime() {
        return (created_time);
    }

    public int getSubsystem() {
        return (subsystem);
    }

    protected void setSubsystem(int _s) {
        subsystem = _s;
    }

    public boolean isAuthorised() {
        return (authorised);
    }

    protected void setAuthorised(boolean _a) {
        authorised = _a;
    }

    public String getPublicKey() {
        return (public_key);
    }

    protected byte[] getRawPublicKey() {
        return (Base32.decode(public_key));
    }

    protected String getShortString() {
        return (public_key.substring(0, 16) + "...");
    }

    public String getNickName() {
        return (nick_name);
    }

    public int getVersion() {
        return (version);
    }

    protected void setVersion(int v) {
        if (version < v) {

            version = v;

            plugin.fireDetailsChanged(this);
        }
    }

    public String getLocalAuthorisedRSSTagsOrCategoriesAsString() {
        synchronized (rss_lock) {

            return (catsToString(rss_local_cats));
        }
    }

    public Set<String> getLocalAuthorisedRSSTagsOrCategories() {
        synchronized (rss_lock) {

            return (rss_local_cats);
        }
    }

    public void addLocalAuthorisedRSSTagOrCategory(String category) {
        category = plugin.normaliseCat(category);

        boolean dirty;

        synchronized (rss_lock) {

            if (rss_local_cats == null) {

                rss_local_cats = new HashSet<String>();
            }

            if (dirty = !rss_local_cats.contains(category)) {

                rss_local_cats.add(category);
            }
        }

        if (dirty) {

            plugin.setConfigDirty();

            plugin.fireDetailsChanged(this);

            // tell buddy of change

            if (isConnected()) {

                sendKeepAlive();
            }
        }
    }

    public void removeLocalAuthorisedRSSTagOrCategory(String category) {
        category = plugin.normaliseCat(category);

        boolean dirty;

        synchronized (rss_lock) {

            if (rss_local_cats == null) {

                return;

            } else {

                dirty = rss_local_cats.remove(category);
            }
        }

        if (dirty) {

            plugin.setConfigDirty();

            plugin.fireDetailsChanged(this);

            // tell buddy of change

            if (isConnected()) {

                sendKeepAlive();
            }
        }
    }

    public void setLocalAuthorisedRSSTagsOrCategories(String new_cats) {
        setLocalAuthorisedRSSTagsOrCategories(stringToCats(new_cats));
    }

    public void setLocalAuthorisedRSSTagsOrCategories(Set<String> new_cats) {
        plugin.normaliseCats(new_cats);

        boolean dirty;

        synchronized (rss_lock) {

            if (dirty = !catsIdentical(new_cats, rss_local_cats)) {

                rss_local_cats = new_cats;
            }
        }

        if (dirty) {

            plugin.setConfigDirty();

            plugin.fireDetailsChanged(this);

            // tell buddy of change

            if (isConnected()) {

                sendKeepAlive();
            }
        }
    }

    public Set<String> getRemoteAuthorisedRSSTagsOrCategories() {
        return (rss_remote_cats);
    }

    public String getRemoteAuthorisedRSSTagsOrCategoriesAsString() {
        return (catsToString(rss_remote_cats));
    }

    protected void setRemoteAuthorisedRSSTagsOrCategories(Set<String> new_cats) {
        plugin.normaliseCats(new_cats);

        boolean dirty;

        synchronized (rss_lock) {

            if (dirty = !catsIdentical(new_cats, rss_remote_cats)) {

                rss_remote_cats = new_cats;
            }
        }

        if (dirty) {

            plugin.setConfigDirty();

            plugin.fireDetailsChanged(this);
        }
    }

    public boolean isLocalRSSTagOrCategoryAuthorised(String category) {
        category = plugin.normaliseCat(category);

        synchronized (rss_lock) {

            if (rss_local_cats != null) {

                return (rss_local_cats.contains(category));
            }

            return (false);
        }
    }

    public boolean isRemoteRSSTagOrCategoryAuthorised(String category) {
        category = plugin.normaliseCat(category);

        synchronized (rss_lock) {

            if (rss_remote_cats != null) {

                return (rss_remote_cats.contains(category));
            }

            return (false);
        }
    }

    protected void localRSSTagOrCategoryRead(String str) {
        boolean dirty;

        synchronized (rss_lock) {

            if (rss_cats_read == null) {

                rss_cats_read = new HashSet<String>();
            }

            dirty = rss_cats_read.add(str);
        }

        if (dirty) {

            // not persisted currently - plugin.setConfigDirty();

            plugin.fireDetailsChanged(this);
        }
    }

    public String getLocalReadTagsOrCategoriesAsString() {
        synchronized (rss_lock) {

            return (catsToString(rss_cats_read));
        }
    }

    public URL getSubscriptionURL(String cat) {
        String url = "azplug:?id=azbuddy&name=Friends&arg=";

        String arg = "pk=" + getPublicKey() + "&cat=" + cat;

        try {
            url += URLEncoder.encode(arg, "UTF-8");

            return (new URL(url));

        } catch (Throwable e) {

            Debug.out(e);

            return (null);
        }
    }

    public void subscribeToCategory(String cat)

    throws BuddyPluginException {
        AZ3Functions.provider az3 = AZ3Functions.getProvider();

        if (az3 == null) {

            throw (new BuddyPluginException("AZ3 subsystem not available"));
        }

        try {
            az3.subscribeToRSS(getName() + ": " + cat, getSubscriptionURL(cat), 15, false, getPublicKey() + ":" + cat);

        } catch (Throwable e) {

            throw (new BuddyPluginException("Failed to add subscription", e));
        }
    }

    public boolean isSubscribedToTagOrCategory(String cat, String creator_ref) {
        if (creator_ref == null) {

            return (false);
        }

        return (creator_ref.equals(getPublicKey() + ":" + cat));
    }

    protected String catsToString(Set<String> cats) {
        if (cats == null || cats.size() == 0) {

            return (null);
        }

        String str = "";

        for (String s : cats) {

            str += (str.length() == 0 ? "" : ",") + s;
        }

        return (str);
    }

    protected boolean catsIdentical(Set<String> c1, Set<String> c2) {
        if (c1 == null && c2 == null) {

            return (true);

        } else if (c1 == null || c2 == null) {

            return (false);

        } else {

            return (c1.equals(c2));
        }
    }

    protected Set<String> stringToCats(String str) {
        if (str == null) {

            return (null);
        }

        String[] bits = str.split(",");

        Set<String> res = new HashSet<String>(bits.length);

        for (String b : bits) {

            b = b.trim();

            if (b.length() > 0) {

                res.add(b);
            }
        }

        if (res.size() == 0) {

            return (null);
        }

        return (res);
    }

    public int getOnlineStatus() {
        return (online_status);
    }

    protected void setOnlineStatus(int s) {
        if (online_status != s) {

            online_status = s;

            plugin.fireDetailsChanged(this);
        }
    }

    public String getName() {
        if (nick_name != null) {

            return (nick_name);
        }

        return (getShortString());
    }

    public void remove() {
        persistent_msg_handler.destroy();

        plugin.removeBuddy(this);
    }

    public InetAddress getIP() {
        return (ip);
    }

    public InetAddress getAdjustedIP() {
        if (ip == null) {

            return (null);
        }

        InetSocketAddress address = new InetSocketAddress(ip, tcp_port);

        InetSocketAddress adjusted_address = AddressUtils.adjustTCPAddress(address, true);

        if (adjusted_address != address) {

            return (adjusted_address.getAddress());
        }

        address = new InetSocketAddress(ip, udp_port);

        adjusted_address = AddressUtils.adjustUDPAddress(address, true);

        if (adjusted_address != address) {

            return (adjusted_address.getAddress());
        }

        return (ip);
    }

    public List getAdjustedIPs() {
        List result = new ArrayList();

        if (ip == null) {

            return (result);
        }

        InetAddress adjusted = getAdjustedIP();

        if (adjusted == ip) {

            result.add(ip);

        } else {

            List l = AddressUtils.getLANAddresses(adjusted.getHostAddress());

            for (int i = 0; i < l.size(); i++) {

                try {
                    result.add(InetAddress.getByName((String) l.get(i)));

                } catch (Throwable e) {

                }
            }
        }

        return (result);
    }

    public int getTCPPort() {
        return (tcp_port);
    }

    public int getUDPPort() {
        return (udp_port);
    }

    public boolean isOnline(boolean is_connected) {
        boolean connected = isConnected();

        // if we're connected then we're online whatever

        if (connected) {

            return (true);
        }

        if (!online) {

            return (false);
        }

        if (is_connected) {

            return (false);

        } else {

            return (true);
        }
    }

    protected boolean isIdle() {
        synchronized (this) {

            return (connections.size() == 0);
        }
    }

    public long getLastTimeOnline() {
        return (last_time_online);
    }

    public BuddyPlugin.cryptoResult encrypt(byte[] payload)

    throws BuddyPluginException {
        return (plugin.encrypt(this, payload));
    }

    public BuddyPlugin.cryptoResult decrypt(byte[] payload)

    throws BuddyPluginException {
        return (plugin.decrypt(this, payload, getName()));
    }

    public boolean verify(byte[] payload, byte[] signature)

    throws BuddyPluginException {

        return (plugin.verify(this, payload, signature));
    }

    public BuddyPluginBuddyMessage storeMessage(int type, Map msg) {
        return (persistent_msg_handler.storeExplicitMessage(type, msg));
    }

    public List<BuddyPluginBuddyMessage> retrieveMessages(int type) {
        return (persistent_msg_handler.retrieveExplicitMessages(type));
    }

    public void setMessagePending()

    throws BuddyPluginException {
        synchronized (this) {

            if (ygm_active) {

                ygm_pending = true;

                return;
            }

            ygm_active = true;
        }

        plugin.setMessagePending(this, new BuddyPlugin.operationListener() {
            public void complete() {
                boolean retry;

                synchronized (BuddyPluginBuddy.this) {

                    ygm_active = false;

                    retry = ygm_pending;

                    ygm_pending = false;
                }

                if (retry) {

                    try {
                        setMessagePending();

                    } catch (BuddyPluginException e) {

                        log("Failed to send YGM", e);
                    }
                }
            }
        });
    }

    public long getLastMessagePending() {
        return (latest_ygm_time);
    }

    protected boolean addYGMMarker(long marker) {
        Long l = new Long(marker);

        synchronized (this) {

            if (recent_ygm == null) {

                recent_ygm = new ArrayList<Long>();
            }

            if (recent_ygm.contains(l)) {

                return (false);
            }

            recent_ygm.add(l);

            if (recent_ygm.size() > 16) {

                recent_ygm.remove(0);
            }

            latest_ygm_time = SystemTime.getCurrentTime();
        }

        plugin.setConfigDirty();

        plugin.fireDetailsChanged(this);

        return (true);
    }

    protected void setLastMessageReceived(String str) {
        last_message_received = str;

        plugin.fireDetailsChanged(this);
    }

    public String getLastMessageReceived() {
        return (last_message_received == null ? "" : last_message_received);
    }

    protected List<Long> getYGMMarkers() {
        synchronized (this) {

            return (recent_ygm == null ? null : new ArrayList<Long>(recent_ygm));
        }
    }

    protected int getLastStatusSeq() {
        synchronized (this) {

            return (last_status_seq);
        }
    }

    protected void buddyConnectionEstablished(boolean outgoing) {
        buddyActive();
    }

    protected void buddyMessageSent(int size, boolean record_active) {
        message_out_count++;
        message_out_bytes += size;

        if (record_active) {

            buddyActive();
        }
    }

    protected void buddyMessageReceived(int size) {
        message_in_count++;
        message_in_bytes += size;

        received_frag_details = "";

        buddyActive();
    }

    protected void buddyMessageFragmentReceived(int num_received, int total) {
        received_frag_details = num_received + "/" + total;

        plugin.fireDetailsChanged(this);
    }

    public String getMessageInFragmentDetails() {
        return (received_frag_details);
    }

    public int getMessageInCount() {
        return (message_in_count);
    }

    public int getMessageOutCount() {
        return (message_out_count);
    }

    public int getBytesInCount() {
        return (message_in_bytes);
    }

    public int getBytesOutCount() {
        return (message_out_bytes);
    }

    public boolean isConnected() {
        boolean connected = false;

        synchronized (this) {

            for (int i = 0; i < connections.size(); i++) {

                buddyConnection c = (buddyConnection) connections.get(i);

                if (c.isConnected() && !c.hasFailed()) {

                    connected = true;
                }
            }
        }

        return (connected);
    }

    protected void buddyActive() {
        long now = SystemTime.getCurrentTime();

        synchronized (this) {

            last_time_online = now;
            online = true;
        }

        persistentDispatchPending();

        plugin.fireDetailsChanged(this);
    }

    public void ping() throws BuddyPluginException {
        plugin.checkAvailable();

        try {
            Map ping_request = new HashMap();

            ping_request.put("type", new Long(BuddyPlugin.RT_INTERNAL_REQUEST_PING));

            sendMessage(BuddyPlugin.SUBSYSTEM_INTERNAL, ping_request, 60 * 1000, new BuddyPluginBuddyReplyListener() {
                public void replyReceived(BuddyPluginBuddy from_buddy, Map reply) {
                    log("Ping reply received:" + reply);
                }

                public void sendFailed(BuddyPluginBuddy to_buddy, BuddyPluginException cause) {
                    log("Ping failed to " + getString(), cause);
                }
            });

        } catch (Throwable e) {

            throw (new BuddyPluginException("Ping failed", e));
        }
    }

    protected void sendCloseRequest(boolean restarting) {
        List to_send = new ArrayList();

        synchronized (this) {

            closing = true;

            for (int i = 0; i < connections.size(); i++) {

                buddyConnection c = (buddyConnection) connections.get(i);

                if (c.isConnected() && !c.hasFailed() && !c.isActive()) {

                    to_send.add(c);
                }
            }
        }

        for (int i = 0; i < to_send.size(); i++) {

            buddyConnection c = (buddyConnection) to_send.get(i);

            try {
                Map close_request = new HashMap();

                close_request.put("type", new Long(BuddyPlugin.RT_INTERNAL_REQUEST_CLOSE));

                close_request.put("r", new Long(restarting ? 1 : 0));

                close_request.put("os", new Long(plugin.getCurrentStatusSeq()));

                final buddyMessage message = new buddyMessage(BuddyPlugin.SUBSYSTEM_INTERNAL, close_request, 60 * 1000);

                message.setListener(new BuddyPluginBuddyReplyListener() {
                    public void replyReceived(BuddyPluginBuddy from_buddy, Map reply) {
                        log("Close reply received:" + reply);
                    }

                    public void sendFailed(BuddyPluginBuddy to_buddy, BuddyPluginException cause) {
                        log("Close failed to " + getString(), cause);
                    }
                });

                c.sendCloseMessage(message);

            } catch (Throwable e) {

                log("Close request failed", e);
            }
        }
    }

    protected void receivedCloseRequest(Map request) {
        List<buddyConnection> closing = new ArrayList<buddyConnection>();

        synchronized (this) {

            closing.addAll(connections);
        }

        for (int i = 0; i < closing.size(); i++) {

            ((buddyConnection) closing.get(i)).remoteClosing();
        }

        try {
            boolean restarting = ((Long) request.get("r")).longValue() == 1;

            if (restarting) {

                logMessage("restarting");

            } else {

                logMessage("going offline");

                boolean details_change = false;

                synchronized (this) {

                    if (offline_seq_set == null) {

                        offline_seq_set = new HashSet<Long>();
                    }

                    offline_seq_set.add(new Long(last_status_seq));

                    offline_seq_set.add((Long) request.get("os"));

                    if (online) {

                        online = false;
                        consec_connect_fails = 0;

                        details_change = true;
                    }
                }

                if (details_change) {

                    plugin.fireDetailsChanged(this);
                }
            }
        } catch (Throwable e) {

            Debug.out("Failed to decode close request", e);
        }
    }

    public void sendMessage(final int subsystem, final Map content, final int timeout_millis, final BuddyPluginBuddyReplyListener listener)

    throws BuddyPluginException {
        plugin.checkAvailable();

        boolean wait = false;

        if (ip == null) {

            synchronized (this) {

                wait = check_active;
            }

            if (!wait) {

                if (SystemTime.getCurrentTime() - last_status_check_time > 30 * 1000) {

                    plugin.updateBuddyStatus(this);

                    wait = true;
                }
            }
        }

        if (wait) {

            new AEThread2("BuddyPluginBuddy:sendWait", true) {
                public void run() {
                    try {
                        long start = SystemTime.getCurrentTime();

                        for (int i = 0; i < 20; i++) {

                            if (ip != null) {

                                break;
                            }

                            Thread.sleep(1000);
                        }

                        long elapsed = SystemTime.getCurrentTime() - start;

                        int new_tm = timeout_millis;

                        if (elapsed > 0 && timeout_millis > 0) {

                            new_tm -= elapsed;

                            if (new_tm <= 0) {

                                listener.sendFailed(BuddyPluginBuddy.this, new BuddyPluginException("Timeout"));

                                return;
                            }
                        }

                        sendMessageSupport(content, subsystem, new_tm, listener);

                    } catch (Throwable e) {

                        if (e instanceof BuddyPluginException) {

                            listener.sendFailed(BuddyPluginBuddy.this, (BuddyPluginException) e);
                        } else {

                            listener.sendFailed(BuddyPluginBuddy.this, new BuddyPluginException("Send failed", e));
                        }
                    }
                }
            }.start();

        } else {

            sendMessageSupport(content, subsystem, timeout_millis, listener);
        }
    }

    protected void sendMessageSupport(final Map content, final int subsystem, final int timeout_millis,
            final BuddyPluginBuddyReplyListener original_listener)

    throws BuddyPluginException {
        boolean too_many_messages = false;

        synchronized (this) {

            too_many_messages = messages.size() >= MAX_QUEUED_MESSAGES;
        }

        if (too_many_messages) {

            throw (new BuddyPluginException("Too many messages queued"));
        }

        final buddyMessage message = new buddyMessage(subsystem, content, timeout_millis);

        BuddyPluginBuddyReplyListener listener_delegate = new BuddyPluginBuddyReplyListener() {
            public void replyReceived(BuddyPluginBuddy from_buddy, Map reply) {
                // logMessage( "Msg " + message.getString() + " ok" );

                try {
                    synchronized (BuddyPluginBuddy.this) {

                        if (current_message != message) {

                            Debug.out("Inconsistent: reply received not for current message");
                        }

                        current_message = null;
                    }

                    original_listener.replyReceived(from_buddy, reply);

                } finally {

                    dispatchMessage();
                }
            }

            public void sendFailed(BuddyPluginBuddy to_buddy, BuddyPluginException cause) {
                logMessage("Msg " + message.getString() + " failed: " + Debug.getNestedExceptionMessage(cause));

                try {
                    // only try and reconcile this failure with the current message if
                    // the message has actually been sent

                    boolean was_active;

                    if (cause instanceof BuddyPluginTimeoutException) {

                        was_active = ((BuddyPluginTimeoutException) cause).wasActive();

                    } else {

                        was_active = true;
                    }

                    if (was_active) {

                        synchronized (BuddyPluginBuddy.this) {

                            if (current_message != message) {

                                Debug.out("Inconsistent: error received not for current message");
                            }

                            current_message = null;
                        }
                    }

                    long now = SystemTime.getCurrentTime();

                    int retry_count = message.getRetryCount();

                    if (retry_count < 1 && !message.timedOut(now)) {

                        message.setRetry();

                        // logMessage( "Msg " + message.getString() + " retrying" );

                        synchronized (BuddyPluginBuddy.this) {

                            messages.add(0, message);
                        }
                    } else {

                        original_listener.sendFailed(to_buddy, cause);
                    }
                } finally {

                    dispatchMessage();
                }
            }
        };

        message.setListener(listener_delegate);

        int size;

        synchronized (this) {

            messages.add(message);

            size = messages.size();
        }

        // logMessage( "Msg " + message.getString() + " added: num=" + size );

        dispatchMessage();
    }

    protected void dispatchMessage() {
        buddyConnection bc = null;

        buddyMessage allocated_message = null;
        Throwable failed_msg_error = null;

        boolean inform_dirty = false;

        synchronized (this) {

            if (current_message != null || messages.size() == 0 || closing) {

                return;
            }

            allocated_message = current_message = (buddyMessage) messages.remove(0);

            for (int i = 0; i < connections.size(); i++) {

                buddyConnection c = (buddyConnection) connections.get(i);

                if (!c.hasFailed()) {

                    bc = c;
                }
            }

            if (bc == null) {

                if (destroyed) {

                    failed_msg_error = new BuddyPluginException("Friend destroyed");

                } else if (connections.size() >= MAX_ACTIVE_CONNECTIONS) {

                    failed_msg_error = new BuddyPluginException("Too many active connections");
                }
            }
        }

        if (failed_msg_error != null) {

            allocated_message.reportFailed(failed_msg_error);

            return;
        }

        if (bc == null) {

            // single-thread outgoing connect attempts

            try {
                outgoing_connect_sem.reserve();

                synchronized (this) {

                    if (current_message != allocated_message) {

                        failed_msg_error = new BuddyPluginException("current message no longer active");

                    } else if (closing) {

                        return;
                    }

                    if (failed_msg_error == null) {

                        for (int i = 0; i < connections.size(); i++) {

                            buddyConnection c = (buddyConnection) connections.get(i);

                            if (!c.hasFailed()) {

                                bc = c;
                            }
                        }

                        if (bc == null) {

                            if (destroyed) {

                                failed_msg_error = new BuddyPluginException("Friend destroyed");

                            } else if (connections.size() >= MAX_ACTIVE_CONNECTIONS) {

                                failed_msg_error = new BuddyPluginException("Too many active connections");
                            }
                        }
                    }
                }

                if (bc == null && failed_msg_error == null) {

                    try {
                        // can't perform connect op while synchronized as may deadlock on password
                        // aquisition

                        GenericMessageConnection generic_connection = outgoingConnection();

                        synchronized (this) {

                            if (current_message != allocated_message) {

                                failed_msg_error = new BuddyPluginException("current message no longer active");

                                generic_connection.close();

                            } else {

                                bc = new buddyConnection(generic_connection, true);

                                inform_dirty = connections.size() == 0;

                                connections.add(bc);

                                // logMessage( "Con " + bc.getString() + " added: num=" + connections.size() );
                            }
                        }
                    } catch (Throwable e) {

                        failed_msg_error = e;
                    }
                }
            } finally {

                outgoing_connect_sem.release();
            }
        }

        if (failed_msg_error != null) {

            allocated_message.reportFailed(failed_msg_error);

            return;
        }

        try {
            // logMessage( "Allocating msg " + allocated_message.getString() + " to con " + bc.getString());

            bc.sendMessage(allocated_message);

        } catch (BuddyPluginException e) {

            allocated_message.reportFailed(e);
        }

        if (inform_dirty) {

            plugin.setConfigDirty();
        }
    }

    protected void removeConnection(buddyConnection bc) {
        int size;

        synchronized (this) {

            connections.remove(bc);

            size = connections.size();
        }

        if (size == 0) {

            plugin.setConfigDirty();
        }

        if (size == 0 && bc.isConnected() && !bc.isClosing() && !bc.isRemoteClosing()) {

            // dropped connection, kick in a keep alive

            if (consec_connect_fails < 3) {

                if (consec_connect_fails == 0) {

                    long now = SystemTime.getMonotonousTime();

                    boolean do_it = false;

                    synchronized (this) {

                        if (last_auto_reconnect == -1 || now - last_auto_reconnect > 30 * 1000) {

                            last_auto_reconnect = now;

                            do_it = true;
                        }
                    }

                    if (do_it) {

                        // delay a bit

                        new DelayedEvent("BuddyPluginBuddy:recon", new Random().nextInt(3000), new AERunnable() {
                            public void runSupport() {
                                int size;

                                synchronized (BuddyPluginBuddy.this) {

                                    size = connections.size();
                                }

                                if (consec_connect_fails == 0 && size == 0) {

                                    log("Attempting reconnect after dropped connection");

                                    sendKeepAlive();
                                }
                            }
                        });
                    }

                } else {

                    long delay = 60 * 1000;

                    delay <<= Math.min(3, consec_connect_fails);

                    if (SystemTime.getCurrentTime() - last_connect_attempt >= delay) {

                        sendKeepAlive();
                    }
                }
            }
        }

        // logMessage( "Con " + bc.getString() + " removed: num=" + size );

        // connection failed, see if we need to attempt to re-establish

        plugin.fireDetailsChanged(this);

        dispatchMessage();
    }

    protected long getLastStatusCheckTime() {
        return (last_status_check_time);
    }

    protected boolean statusCheckActive() {
        synchronized (this) {

            return (check_active);
        }
    }

    protected boolean statusCheckStarts() {
        synchronized (this) {

            if (check_active) {

                return (false);
            }

            last_status_check_time = SystemTime.getCurrentTime();

            check_active = true;
        }

        return (true);
    }

    protected void statusCheckFailed() {
        boolean details_change = false;

        synchronized (this) {

            try {
                if (online) {

                    online = false;
                    consec_connect_fails = 0;

                    details_change = true;
                }
            } finally {

                status_check_count++;

                check_active = false;
            }
        }

        if (details_change) {

            plugin.fireDetailsChanged(this);
        }
    }

    protected void setCachedStatus(InetAddress _ip, int _tcp_port, int _udp_port) {
        synchronized (this) {

            if (ip == null) {

                ip = _ip;
                tcp_port = _tcp_port;
                udp_port = _udp_port;
            }
        }
    }

    protected void statusCheckComplete(long _post_time, InetAddress _ip, int _tcp_port, int _udp_port, String _nick_name, int _online_status,
            int _status_seq, int _version) {
        boolean details_change = false;
        boolean config_dirty = false;

        long now = SystemTime.getCurrentTime();

        if (now < last_time_online) {

            last_time_online = now;
        }

        boolean is_connected = isConnected();

        synchronized (this) {

            try {
                // do we explicitly know that this sequence number denotes an offline buddy

                if (offline_seq_set != null) {

                    if (offline_seq_set.contains(new Long(_status_seq))) {

                        return;

                    } else {

                        offline_seq_set = null;
                    }
                }

                boolean seq_change = _status_seq != last_status_seq;

                boolean timed_out;

                // change in sequence means we're online

                if (seq_change) {

                    last_status_seq = _status_seq;
                    last_time_online = now;

                    timed_out = false;
                    details_change = true;

                } else {

                    timed_out = now - last_time_online >= BuddyPlugin.STATUS_REPUBLISH_PERIOD * 3;
                }

                if (online) {

                    if (timed_out) {

                        online = false;
                        consec_connect_fails = 0;

                        details_change = true;
                    }
                } else {

                    if (seq_change || !timed_out) {

                        online = true;
                        details_change = true;
                    }
                }

                post_time = _post_time;

                if (!addressesEqual(ip, _ip) || tcp_port != _tcp_port || udp_port != _udp_port || version < _version) {

                    ip = _ip;
                    tcp_port = _tcp_port;
                    udp_port = _udp_port;

                    if (version < _version) {

                        version = _version;
                    }

                    details_change = true;
                }

                // if we are connected then we use the status sent over the connection
                // as it is more up to date

                if (!is_connected && online_status != _online_status) {

                    online_status = _online_status;

                    details_change = true;
                }

                if (!plugin.stringsEqual(nick_name, _nick_name)) {

                    nick_name = _nick_name;

                    config_dirty = true;
                    details_change = true;
                }
            } finally {

                status_check_count++;

                check_active = false;
            }
        }

        if (config_dirty) {

            plugin.setConfigDirty();
        }

        if (details_change) {

            if (online) {

                persistentDispatchPending();
            }

            plugin.fireDetailsChanged(this);
        }

        plugin.logMessage(getString());
    }

    protected boolean addressesEqual(InetAddress ip1, InetAddress ip2) {
        if (ip1 == null && ip2 == null) {

            return (true);

        } else if (ip1 == null || ip2 == null) {

            return (false);

        } else {

            return (ip1.equals(ip2));
        }
    }

    protected void checkTimeouts() {
        long now = SystemTime.getCurrentTime();

        List failed = null;

        List connections_to_check = null;

        boolean messages_queued;

        synchronized (this) {

            messages_queued = messages.size() > 0;

            if (messages_queued) {

                Iterator it = messages.iterator();

                while (it.hasNext()) {

                    buddyMessage message = (buddyMessage) it.next();

                    if (message.timedOut(now)) {

                        it.remove();

                        if (failed == null) {

                            failed = new ArrayList();
                        }

                        failed.add(message);
                    }
                }
            }

            if (connections.size() > 0) {

                connections_to_check = new ArrayList(connections);
            }
        }

        boolean send_keep_alive = false;

        if (connections_to_check == null) {

            // no active connections

            if (online && ip != null && !messages_queued) {

                // see if we should attempt a pre-emptive connect

                if (consec_connect_fails < 3) {

                    long delay = 60 * 1000;

                    delay <<= Math.min(3, consec_connect_fails);

                    send_keep_alive = now - last_connect_attempt >= delay;
                }
            }
        } else {

            for (int i = 0; i < connections_to_check.size(); i++) {

                buddyConnection connection = (buddyConnection) connections_to_check.get(i);

                boolean closed = connection.checkTimeout(now);

                if (ip != null && !closed && !messages_queued && connection.isConnected() && !connection.isActive()) {

                    if (now - connection.getLastActive(now) > CONNECTION_KEEP_ALIVE) {

                        send_keep_alive = true;
                    }
                }
            }
        }

        if (send_keep_alive) {

            sendKeepAlive();
        }

        if (failed != null) {

            for (int i = 0; i < failed.size(); i++) {

                ((buddyMessage) failed.get(i)).reportFailed(new BuddyPluginTimeoutException("Timeout", false));
            }
        }
    }

    protected void sendKeepAlive() {
        boolean send_keep_alive = true;

        synchronized (this) {

            if (keep_alive_outstanding) {

                send_keep_alive = false;

            } else {

                keep_alive_outstanding = true;
            }
        }

        if (send_keep_alive) {

            try {
                Map ping_request = new HashMap();

                ping_request.put("type", new Long(BuddyPlugin.RT_INTERNAL_REQUEST_PING));

                sendMessageSupport(ping_request, BuddyPlugin.SUBSYSTEM_INTERNAL, 60 * 1000, new BuddyPluginBuddyReplyListener() {
                    public void replyReceived(BuddyPluginBuddy from_buddy, Map reply) {
                        synchronized (BuddyPluginBuddy.this) {

                            keep_alive_outstanding = false;
                        }
                    }

                    public void sendFailed(BuddyPluginBuddy to_buddy, BuddyPluginException cause) {
                        synchronized (BuddyPluginBuddy.this) {

                            keep_alive_outstanding = false;
                        }
                    }
                });

            } catch (Throwable e) {

                synchronized (this) {

                    keep_alive_outstanding = false;
                }
            }
        }
    }

    public String getConnectionsString() {
        synchronized (this) {

            String str = "";

            for (int i = 0; i < connections.size(); i++) {

                str += (str.length() == 0 ? "" : ",") + ((buddyConnection) connections.get(i)).getString(true);
            }

            return (str);
        }
    }

    public void disconnect() {
        List to_disconnect = new ArrayList();

        synchronized (this) {

            to_disconnect.addAll(connections);
        }

        for (int i = 0; i < to_disconnect.size(); i++) {

            ((buddyConnection) to_disconnect.get(i)).disconnect();
        }
    }

    protected boolean isClosing() {
        return (closing);
    }

    protected void destroy() {
        List<buddyConnection> to_close = new ArrayList<buddyConnection>();

        synchronized (this) {

            destroyed = true;

            to_close.addAll(connections);
        }

        for (int i = 0; i < to_close.size(); i++) {

            ((buddyConnection) to_close.get(i)).close();
        }
    }

    protected void logMessage(String str) {
        plugin.logMessage(getShortString() + ": " + str);
    }

    protected GenericMessageConnection outgoingConnection()

    throws BuddyPluginException {
        GenericMessageRegistration msg_registration = plugin.getMessageRegistration();

        if (msg_registration == null) {

            throw (new BuddyPluginException("Messaging system unavailable"));
        }

        InetAddress ip = getIP();

        if (ip == null) {

            throw (new BuddyPluginException("Friend offline (no usable IP address)"));
        }

        InetSocketAddress tcp_target = null;
        InetSocketAddress udp_target = null;

        int tcp_port = getTCPPort();

        if (tcp_port > 0) {

            tcp_target = new InetSocketAddress(ip, tcp_port);
        }

        int udp_port = getUDPPort();

        if (udp_port > 0) {

            udp_target = new InetSocketAddress(ip, udp_port);
        }

        InetSocketAddress notional_target = tcp_target;

        if (notional_target == null) {

            notional_target = udp_target;
        }

        if (notional_target == null) {

            throw (new BuddyPluginException("Friend offline (no usable protocols)"));
        }

        GenericMessageEndpoint endpoint = msg_registration.createEndpoint(notional_target);

        if (tcp_target != null) {

            endpoint.addTCP(tcp_target);
        }

        if (udp_target != null) {

            endpoint.addUDP(udp_target);
        }

        GenericMessageConnection con = null;

        try {
            last_connect_attempt = SystemTime.getCurrentTime();

            con = msg_registration.createConnection(endpoint);

            plugin.addRateLimiters(con);

            String reason = "Friend: Outgoing connection establishment";

            SESecurityManager sec_man = plugin.getSecurityManager();

            con = sec_man.getSTSConnection(con, sec_man.getPublicKey(SEPublicKey.KEY_TYPE_ECC_192, reason),

            new SEPublicKeyLocator() {
                public boolean accept(Object context, SEPublicKey other_key) {
                    String other_key_str = Base32.encode(other_key.encodeRawPublicKey());

                    if (other_key_str.equals(public_key)) {

                        consec_connect_fails = 0;

                        return (true);

                    } else {

                        log(getString() + ": connection failed due to pk mismatch");

                        return (false);
                    }
                }
            }, reason, BuddyPlugin.BLOCK_CRYPTO);

            con.connect();

            return (con);

        } catch (Throwable e) {

            if (con != null) {

                consec_connect_fails++;

                try {
                    con.close();

                } catch (Throwable f) {

                    log("Failed to close connection", f);
                }
            }

            throw (new BuddyPluginException("Failed to send message", e));
        }
    }

    protected void incomingConnection(GenericMessageConnection _connection)

    throws BuddyPluginException {
        addConnection(_connection);
    }

    protected void addConnection(GenericMessageConnection _connection)

    throws BuddyPluginException {
        // int size;

        buddyConnection bc = new buddyConnection(_connection, false);

        boolean inform_dirty = false;

        synchronized (this) {

            if (destroyed) {

                throw (new BuddyPluginException("Friend has been destroyed"));
            }

            inform_dirty = connections.size() == 0;

            connections.add(bc);

            // size = connections.size();
        }

        // logMessage( "Con " + bc.getString() + " added: num=" + size );

        if (inform_dirty) {

            plugin.setConfigDirty();
        }
    }

    public void setUserData(Object key, Object value) {
        synchronized (user_data) {

            user_data.put(key, value);
        }
    }

    public Object getUserData(Object key) {
        synchronized (user_data) {

            return (user_data.get(key));
        }
    }

    protected void log(String str) {
        plugin.log(str);
    }

    protected void log(String str, Throwable e) {
        plugin.log(str, e);
    }

    public String getString() {
        return ("pk=" + getShortString() + (nick_name == null ? "" : (",nick=" + nick_name)) + ",ip=" + ip + ",tcp=" + tcp_port + ",udp="
                + udp_port + ",online=" + online + ",age=" + (SystemTime.getCurrentTime() - post_time));
    }

    protected class buddyMessage {
        private int message_id;

        private Map request;
        private int subsystem;
        private BuddyPluginBuddyReplyListener listener;
        private int timeout_millis;

        private long queue_time = SystemTime.getCurrentTime();

        private boolean timed_out;
        private int retry_count;
        private boolean complete;

        protected buddyMessage(int _subsystem, Map _request, int _timeout) {
            synchronized (BuddyPluginBuddy.this) {

                message_id = next_message_id++;
            }

            request = _request;
            subsystem = _subsystem;
            timeout_millis = _timeout;
        }

        protected void setListener(BuddyPluginBuddyReplyListener _listener) {
            listener = _listener;
        }

        protected int getRetryCount() {
            synchronized (this) {

                return (retry_count);
            }
        }

        protected void setDontRetry() {
            retry_count = 99;
        }

        protected void setRetry() {
            synchronized (this) {

                retry_count++;

                complete = false;
                timed_out = false;

            }
        }

        protected boolean timedOut(long now) {
            if (timed_out) {

                return (true);
            }

            if (now < queue_time) {

                queue_time = now;

                return (false);

            } else {

                timed_out = now - queue_time >= timeout_millis;

                return (timed_out);
            }
        }

        protected Map getRequest() {
            return (request);
        }

        protected int getSubsystem() {
            return (subsystem);
        }

        protected int getID() {
            return (message_id);
        }

        protected void reportComplete(Map reply) {
            synchronized (this) {

                if (complete) {

                    return;
                }

                complete = true;
            }

            try {
                listener.replyReceived(BuddyPluginBuddy.this, reply);

            } catch (Throwable e) {

                Debug.printStackTrace(e);
            }
        }

        protected void reportFailed(Throwable error) {
            synchronized (this) {

                if (complete) {

                    return;
                }

                complete = true;
            }

            try {
                if (error instanceof BuddyPluginException) {

                    listener.sendFailed(BuddyPluginBuddy.this, (BuddyPluginException) error);

                } else {

                    listener.sendFailed(BuddyPluginBuddy.this, new BuddyPluginException("", error));
                }
            } catch (Throwable e) {

                Debug.printStackTrace(e);
            }
        }

        protected String getString() {
            return ("id=" + message_id + ",ss=" + subsystem + (retry_count == 0 ? "" : (",retry=" + retry_count)));
        }
    }

    protected class buddyConnection implements fragmentHandlerReceiver {
        private fragmentHandler fragment_handler;
        private int connection_id;
        private boolean outgoing;

        private String dir_str;

        private volatile buddyMessage active_message;

        private volatile boolean connected;
        private volatile boolean closing;
        private volatile boolean remote_closing;
        private volatile boolean failed;

        private long last_active = SystemTime.getCurrentTime();

        protected buddyConnection(GenericMessageConnection _connection, boolean _outgoing) {
            fragment_handler = new fragmentHandler(_connection, this);

            outgoing = _outgoing;

            synchronized (BuddyPluginBuddy.this) {

                connection_id = next_connection_id++;
            }

            dir_str = outgoing ? "Outgoing" : "Incoming";

            if (!outgoing) {

                connected = true;

                buddyConnectionEstablished(false);
            }

            fragment_handler.start();
        }

        protected boolean isConnected() {
            return (connected);
        }

        protected boolean hasFailed() {
            return (failed);
        }

        protected boolean isOutgoing() {
            return (outgoing);
        }

        protected long getLastActive(long now) {
            if (now < last_active) {

                last_active = now;
            }

            return (last_active);
        }

        protected void sendMessage(buddyMessage message)

        throws BuddyPluginException {
            BuddyPluginException failed_error = null;

            buddyMessage msg_to_send = null;

            synchronized (this) {

                if (BuddyPluginBuddy.this.isClosing()) {

                    throw (new BuddyPluginException("Close in progress"));
                }

                if (active_message != null) {

                    Debug.out("Inconsistent: active message already set");

                    failed_error = new BuddyPluginException("Inconsistent state");

                } else if (failed || closing) {

                    throw (new BuddyPluginException("Connection failed"));

                } else {

                    active_message = message;

                    if (connected) {

                        msg_to_send = active_message;
                    }
                }
            }

            if (failed_error != null) {

                failed(failed_error);

                throw (failed_error);
            }

            if (msg_to_send != null) {

                send(msg_to_send);
            }
        }

        protected void sendCloseMessage(buddyMessage message) {
            boolean ok_to_send;

            synchronized (this) {

                ok_to_send = active_message == null && connected && !failed && !closing;
            }

            if (ok_to_send) {

                send(message);
            }
        }

        public boolean isActive() {
            return (active_message != null);
        }

        public void connected() {
            if (TRACE) {
                System.out.println(dir_str + " connected");
            }

            buddyMessage msg_to_send = null;

            synchronized (this) {

                last_active = SystemTime.getCurrentTime();

                connected = true;

                msg_to_send = active_message;
            }

            buddyConnectionEstablished(true);

            if (msg_to_send != null) {

                send(msg_to_send);
            }
        }

        protected boolean checkTimeout(long now) {
            buddyMessage bm = null;

            boolean close = false;

            synchronized (this) {

                if (active_message != null) {

                    if (active_message.timedOut(now)) {

                        bm = active_message;

                        active_message = null;
                    }
                }

                if (now < last_active) {

                    last_active = now;
                }

                if (now - last_active > CONNECTION_IDLE_TIMEOUT) {

                    close = true;
                }
            }

            if (bm != null) {

                bm.reportFailed(new BuddyPluginTimeoutException("Timeout", true));
            }

            if (close) {

                close();
            }

            return (close);
        }

        protected void send(buddyMessage msg) {
            Map request = msg.getRequest();

            Map send_map = new HashMap();

            send_map.put("type", new Long(RT_REQUEST_DATA));
            send_map.put("req", request);
            send_map.put("ss", new Long(msg.getSubsystem()));
            send_map.put("id", new Long(msg.getID()));
            send_map.put("oz", new Long(plugin.getOnlineStatus()));
            send_map.put("v", new Long(BuddyPlugin.VERSION_CURRENT));

            String loc_cat = getLocalAuthorisedRSSTagsOrCategoriesAsString();

            if (loc_cat != null) {
                send_map.put("cat", loc_cat);
            }

            try {
                // logMessage( "Sending " + msg.getString() + " to " + getString());

                fragment_handler.send(send_map, true, true);

                synchronized (this) {

                    last_active = SystemTime.getCurrentTime();
                    ;
                }
            } catch (BuddyPluginException e) {

                try {
                    failed(e);

                } catch (Throwable f) {

                    Debug.printStackTrace(f);
                }
            }
        }

        public void receive(Map data_map) {
            synchronized (this) {

                last_active = SystemTime.getCurrentTime();
                ;
            }

            if (TRACE) {
                System.out.println(dir_str + " receive: " + data_map);
            }

            try {
                int type = ((Long) data_map.get("type")).intValue();

                Long l_os = (Long) data_map.get("oz");

                if (l_os != null) {

                    setOnlineStatus(l_os.intValue());
                }

                Long l_ver = (Long) data_map.get("v");

                if (l_ver != null) {

                    setVersion(l_ver.intValue());
                }

                byte[] b_rem_cat = (byte[]) data_map.get("cat");

                if (b_rem_cat == null) {

                    setRemoteAuthorisedRSSTagsOrCategories(null);

                } else {

                    setRemoteAuthorisedRSSTagsOrCategories(stringToCats(new String(b_rem_cat, "UTF-8")));
                }

                if (type == RT_REQUEST_DATA) {

                    // logMessage( "Received type=" + type + " from " + getString());

                    Long subsystem = (Long) data_map.get("ss");

                    Map reply;

                    int reply_type;

                    Map request = (Map) data_map.get("req");

                    String error = null;

                    if (request == null || subsystem == null) {

                        reply = null;

                    } else {

                        try {

                            reply = plugin.requestReceived(BuddyPluginBuddy.this, subsystem.intValue(), request);

                        } catch (Throwable e) {

                            error = Debug.getNestedExceptionMessage(e);

                            reply = null;
                        }
                    }

                    if (reply == null) {

                        reply_type = RT_REPLY_ERROR;

                        reply = new HashMap();

                        reply.put("error", error == null ? "No handlers available to process request" : error);

                    } else {

                        reply_type = RT_REPLY_DATA;
                    }

                    Map reply_map = new HashMap();

                    reply_map.put("ss", subsystem);
                    reply_map.put("type", new Long(reply_type));
                    reply_map.put("id", data_map.get("id"));
                    reply_map.put("oz", new Long(plugin.getOnlineStatus()));

                    String loc_cat = getLocalAuthorisedRSSTagsOrCategoriesAsString();

                    if (loc_cat != null) {
                        reply_map.put("cat", loc_cat);
                    }

                    reply_map.put("rep", reply);

                    // don't record as active here as (1) we recorded as active above when
                    // receiving request (2) we may be replying to a 'closing' message and
                    // we don't want the reply to mark as online

                    fragment_handler.send(reply_map, false, false);

                } else if (type == RT_REPLY_DATA || type == RT_REPLY_ERROR) {

                    long id = ((Long) data_map.get("id")).longValue();

                    buddyMessage bm;

                    synchronized (this) {

                        if (active_message != null && active_message.getID() == id) {

                            bm = active_message;

                            active_message = null;

                        } else {

                            bm = null;
                        }
                    }

                    Map reply = (Map) data_map.get("rep");

                    if (bm == null) {

                        logMessage("reply discarded as no matching request: " + reply);

                    } else {

                        if (type == RT_REPLY_ERROR) {

                            bm.setDontRetry();

                            bm.reportFailed(new BuddyPluginException(new String((byte[]) reply.get("error"))));

                        } else {

                            bm.reportComplete(reply);
                        }
                    }
                } else {

                    // ignore unknown message types
                }
            } catch (Throwable e) {

                failed(e);
            }
        }

        protected void close() {
            closing = true;

            failed(new BuddyPluginException("Closing"));
        }

        protected boolean isClosing() {
            return (closing);
        }

        protected void remoteClosing() {
            remote_closing = true;
        }

        protected boolean isRemoteClosing() {
            return (remote_closing);
        }

        protected void disconnect() {
            fragment_handler.close();
        }

        public void failed(Throwable error) {
            buddyMessage bm = null;

            if (!connected && outgoing) {

                consec_connect_fails++;
            }

            synchronized (this) {

                if (failed) {

                    return;
                }

                failed = true;

                bm = active_message;

                active_message = null;
            }

            logMessage("Con " + getString() + " failed: " + Debug.getNestedExceptionMessage(error));

            try {
                if (!closing) {

                    if (TRACE) {
                        System.out.println(dir_str + " connection error:");
                    }
                }

                fragment_handler.close();

            } finally {

                removeConnection(this);

                if (bm != null) {

                    bm.reportFailed(error);
                }
            }
        }

        protected String getString() {
            return (getString(false));
        }

        protected String getString(boolean short_form) {
            if (short_form) {

                return (fragment_handler.getString());

            } else {

                return ("id=" + connection_id + ",dir=" + (outgoing ? "out" : "in"));
            }
        }
    }

    protected class fragmentHandler implements GenericMessageConnectionListener {
        private GenericMessageConnection connection;
        private fragmentHandlerReceiver receiver;

        private int next_fragment_id = 0;

        private fragmentAssembly current_request_frag;
        private fragmentAssembly current_reply_frag;

        private int send_count;
        private int recv_count;

        protected fragmentHandler(GenericMessageConnection _connection, fragmentHandlerReceiver _receiver) {
            connection = _connection;
            receiver = _receiver;
        }

        public void start() {
            connection.addListener(this);
        }

        public void connected(GenericMessageConnection connection) {
            receiver.connected();
        }

        public void failed(GenericMessageConnection connection, Throwable error)

        throws MessageException {
            receiver.failed(error);
        }

        protected void send(Map data_map, boolean is_request, boolean record_active)

        throws BuddyPluginException {
            try {
                byte[] data = BEncoder.encode(data_map);

                int data_length = data.length;

                plugin.checkMaxMessageSize(data_length);

                int max_chunk = connection.getMaximumMessageSize() - 1024;

                if (data_length > max_chunk) {

                    int fragment_id;

                    synchronized (this) {

                        fragment_id = next_fragment_id++;
                    }

                    int chunk_num = 0;

                    for (int i = 0; i < data_length; i += max_chunk) {

                        int end = Math.min(data_length, i + max_chunk);

                        if (end > i) {

                            byte[] chunk = new byte[end - i];

                            System.arraycopy(data, i, chunk, 0, chunk.length);

                            Map chunk_map = new HashMap();

                            chunk_map.put("type", new Long(BuddyPlugin.RT_INTERNAL_FRAGMENT));
                            chunk_map.put("f", new Long(fragment_id));
                            chunk_map.put("l", new Long(data_length));
                            chunk_map.put("c", new Long(max_chunk));
                            chunk_map.put("i", new Long(chunk_num));
                            chunk_map.put("q", new Long(is_request ? 1 : 0));
                            chunk_map.put("d", chunk);

                            byte[] chunk_data = BEncoder.encode(chunk_map);

                            PooledByteBuffer chunk_buffer = plugin.getPluginInterface().getUtilities().allocatePooledByteBuffer(chunk_data);

                            try {
                                connection.send(chunk_buffer);

                                chunk_buffer = null;

                            } finally {

                                if (chunk_buffer != null) {

                                    chunk_buffer.returnToPool();
                                }
                            }
                        }

                        chunk_num++;
                    }
                } else {

                    PooledByteBuffer buffer = plugin.getPluginInterface().getUtilities().allocatePooledByteBuffer(data);

                    try {

                        connection.send(buffer);

                        buffer = null;

                    } finally {

                        if (buffer != null) {

                            buffer.returnToPool();
                        }
                    }
                }

                buddyMessageSent(data.length, record_active);

                send_count++;

            } catch (Throwable e) {

                throw (new BuddyPluginException("Send failed", e));
            }
        }

        public void receive(GenericMessageConnection connection, PooledByteBuffer message)

        throws MessageException {
            try {
                // while in unauth state we only allow a few messages. max should be 1
                // for an 'accept request' but I feel generous

                if (recv_count >= 4 && !isAuthorised()) {

                    throw (new MessageException("Too many messages received while in unauthorised state"));
                }

                byte[] content = message.toByteArray();

                Map data_map = BDecoder.decode(content);

                if (((Long) data_map.get("type")).intValue() == BuddyPlugin.RT_INTERNAL_FRAGMENT) {

                    Map chunk_map = data_map;

                    int fragment_id = ((Long) chunk_map.get("f")).intValue();
                    int data_length = ((Long) chunk_map.get("l")).intValue();
                    int chunk_size = ((Long) chunk_map.get("c")).intValue();
                    int chunk_num = ((Long) chunk_map.get("i")).intValue();

                    boolean is_request = ((Long) chunk_map.get("q")).intValue() == 1;

                    byte[] chunk_data = (byte[]) chunk_map.get("d");

                    plugin.checkMaxMessageSize(data_length);

                    fragmentAssembly assembly;

                    if (is_request) {

                        if (current_request_frag == null) {

                            current_request_frag = new fragmentAssembly(fragment_id, data_length, chunk_size);
                        }

                        assembly = current_request_frag;

                    } else {

                        if (current_reply_frag == null) {

                            current_reply_frag = new fragmentAssembly(fragment_id, data_length, chunk_size);
                        }

                        assembly = current_reply_frag;
                    }

                    if (assembly.getID() != fragment_id) {

                        throw (new BuddyPluginException("Fragment receive error: concurrent decode not supported"));
                    }

                    if (assembly.receive(chunk_num, chunk_data)) {

                        if (is_request) {

                            current_request_frag = null;

                        } else {

                            current_reply_frag = null;
                        }

                        buddyMessageReceived(data_length);

                        recv_count++;

                        receiver.receive(BDecoder.decode(assembly.getData()));

                    } else {

                        buddyMessageFragmentReceived(assembly.getChunksReceived(), assembly.getTotalChunks());
                    }
                } else {

                    buddyMessageReceived(content.length);

                    recv_count++;

                    receiver.receive(data_map);
                }
            } catch (Throwable e) {

                receiver.failed(e);

            } finally {

                message.returnToPool();
            }
        }

        protected void close() {
            try {

                connection.close();

            } catch (Throwable e) {

                // Debug.printStackTrace( e );

            } finally {

                receiver.failed(new Exception("Connection closed"));
            }
        }

        protected String getString() {
            return (connection.getType());
        }

        protected class fragmentAssembly {
            private int id;
            private byte[] data;
            private int chunk_size;

            private int num_chunks;
            private Set chunks_received = new HashSet();

            protected fragmentAssembly(int _id, int _length, int _chunk_size) {
                id = _id;
                chunk_size = _chunk_size;

                data = new byte[_length];

                num_chunks = (_length + chunk_size - 1) / chunk_size;
            }

            protected int getID() {
                return (id);
            }

            protected int getChunksReceived() {
                return (chunks_received.size());
            }

            protected int getTotalChunks() {
                return (num_chunks);
            }

            protected boolean receive(int chunk_num, byte[] chunk) {
                // System.out.println( "received chunk " + chunk_num + " of " + num_chunks );

                Integer i = new Integer(chunk_num);

                if (chunks_received.contains(i)) {

                    return (false);
                }

                chunks_received.add(i);

                System.arraycopy(chunk, 0, data, chunk_num * chunk_size, chunk.length);

                return (chunks_received.size() == num_chunks);
            }

            protected byte[] getData() {
                return (data);
            }
        }
    }

    interface fragmentHandlerReceiver {
        public void connected();

        public void receive(Map data);

        public void failed(Throwable error);
    }
}
